package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.util.Random

object Spark_stream_DIY {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStream")
    val ssc = new StreamingContext(sparkConf, Seconds(3) )
    ssc.sparkContext.setLogLevel("ERROR")

    //创建自定义 receiver的Streaming
    val messageDS: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver())
    messageDS.print()

    ssc.start()
    ssc.awaitTermination()//等待数据采集--阻塞主线程  Thread
  }
  /*
    自定义数据采集器
    Spark Streaming提供很多采集器 Kafak Flume HDFS, 唯独没有 MySQL
    1.继承Receiver。定义泛型，传递参数
    2.重写方法 onStart onStop
   */
  class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY){

    private var flag = true
    //启动的时候，就调用该方法，采集数据并将数据返回给Spark
    override def onStart(): Unit = {
      new Thread(new Runnable {
        override def run(): Unit = {
          while (flag){
            //连接MySQL,读取某张数据表，进行数据采集
            val message = "采集的数据为："+new Random().nextInt(10).toString

            store(message) //不能用return返回，因为带有结束方法的作用。
            Thread.sleep(2000)

          }
        }
      }).start()
    }

    override def onStop(): Unit = {
      flag = false
    }
  }
}
